package com.bw.yk10.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.yk10.func.HbaseSinkFunction;
import com.bw.yk10.util.DimHbaseSink;
import com.bw.yk10.util.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Text02 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        4、公共维度层DIM：编写Flink 流式程序，实时消费ODS层线上问诊业务数据，提取出核心维表数据（字典表dict、医院信息表hospital、医生信息表doctor、就诊人信息表patient），并对json数据解析过滤，实时存储业务数据到HBase表（表名称前缀：dim_，表列簇名称：info）

        DataStreamSource<String> kafkaDS= env.addSource(new MyKafkaUtil().getKafkaConsumer("ods-medical-topic","test02_group12"));

        OutputTag<String> dictOutputTag = new OutputTag<String>("dict") {};
        OutputTag<String> hospitalOutputTag = new OutputTag<String>("hospital") {};
        OutputTag<String> doctorOutputTag = new OutputTag<String>("doctor") {};
        OutputTag<String> patientOutputTag = new OutputTag<String>("patient") {};

        SingleOutputStreamOperator<String> dataDS = kafkaDS.process(new ProcessFunction<String,String>() {
            @Override
            public void processElement(String o, Context context, Collector<String> collector) throws Exception {

                JSONObject jsonObject = JSON.parseObject(o);
                if(jsonObject.getString("table").equals("dict")) {
                    context.output(dictOutputTag, o);
                }

                if(jsonObject.getString("table").equals("hospital")) {
                    context.output(hospitalOutputTag, o);
                }

                if(jsonObject.getString("table").equals("doctor")) {
                    context.output(doctorOutputTag, o);
                }

                if(jsonObject.getString("table").equals("patient")) {
                    context.output(patientOutputTag, o);
                }

                if(jsonObject.getString("table").equals("consultation")) {
                    collector.collect(o);
                }
            }
        });

        DataStream<String> dictDS=dataDS.getSideOutput(dictOutputTag);
        DataStream<String> hospitalDS=dataDS.getSideOutput(hospitalOutputTag);
        DataStream<String> doctorDS=dataDS.getSideOutput(doctorOutputTag);
        DataStream<String> patientDS=dataDS.getSideOutput(patientOutputTag);

        dictDS.addSink(new DimHbaseSink());
        hospitalDS.addSink(new DimHbaseSink());
        doctorDS.addSink(new DimHbaseSink());
        patientDS.addSink(new DimHbaseSink());

//        dataDS.addSink(new HbaseSinkFunction());
        dictDS.print("dictDS=======>");
        hospitalDS.print("hospitalDS=======>");
        doctorDS.print("doctorDS=======>");
        patientDS.print("patientDS=======>");


        //dwd-trade-consultation
//        dataDS.print();

        dataDS.addSink(new MyKafkaUtil().getKafkaProducer("dwd-trade-consultation"));
        env.execute();
    }
}
